package cn.getech.data.development.utils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.permission.FsPermission
import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}
import org.slf4j.{Logger, LoggerFactory}

object HDFSUtils {
  System.setProperty(ConfigurationManager.getProperty("hdfs.account"), ConfigurationManager.getProperty("hdfs.username"))
  private val logger: Logger = LoggerFactory.getLogger(this.getClass)

  /**
   * 通用
   * @param path 路径
   * @param bk block
   */
  def block(path: String, bk: (FileSystem, Path) => Unit): Unit = {
    val fileOutPath = new Path(path)
    var system: FileSystem = null
    try {
      system = FileSystem.get(new Configuration)
      bk(system, fileOutPath)
    } catch {
      case e: Exception =>
        if (system != null)
          system.close()
        logger.error(e.getMessage)
    }
  }

  /**
   * 创建文件
   * @param path 路径
   * @param permission FsPermission 权限
   */
  def mkdirs(path: String, permission: FsPermission = FsPermission.getDefault): Unit = {
    block(path, { (sys, path) =>
      if (!sys.exists(path)) {
        sys.mkdirs(path, permission)
      }
    })
  }

  /**
   * 删除
   * @param path 路径
   */
  def delete(path: String) = {
    block(path, { (sys, path) =>
      if (sys.exists(path)) {
        sys.delete(path, true)
      }
    })
  }

  /**
   * 重命名
   * @param path 路径
   * @param targetPath 目标路径
   */
  def rename(path: String, targetPath: String) = {
    block(path, { (sys, path) =>
      if (sys.exists(path)) {
        sys.rename(path, new Path(targetPath))
      }
    })
  }

  /**
   * 迁移数据
   * @param path 路径
   * @param targetFileName 目标路径文件名
   * @param overrideName  迁移数据的文件名
   */
  def overridePath(path: String, targetFileName: String, overrideName: String = "/.before"): Unit = {
    var tmpPath = path
    if (path(path.length - 1) != '/') tmpPath = path + '/'
    val targetPath = new Path(tmpPath + targetFileName + overrideName)

    logger.info("override data path: ", tmpPath + targetFileName + overrideName)
    block(tmpPath + targetFileName, { (sys, curPath) =>
      val value = sys.listStatus(curPath)
      val paths = FileUtil.stat2Paths(value)
      if (paths.length > 0 && !sys.exists(targetPath)) sys.mkdirs(targetPath)
      paths.foreach(sys.rename(_, targetPath))
      logger.info("Overwrite operation completed!")
    })
  }

}
